package com.enba.boot.mq.sender.kafka;

import com.enba.boot.mq.MessageInteface;
import com.enba.boot.mq.MessageSenderStrategy;
import com.enba.boot.mq.properties.MqProperties;
import com.enba.boot.mq.sender.activemq.ActiveMqMessageBase;
import javax.annotation.Resource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

/** kafka 集群模式消息发送 */
public class KafkaClusterMessageSenderStrategy
    implements MessageSenderStrategy<ActiveMqMessageBase> {

  @Resource private KafkaTemplate<Object, Object> kafkaTemplate;

  @Autowired private MqProperties mqProperties;

  @Override
  public String getStrategyKey() {
    return mqProperties.getKafka().getStrategyKey();
  }

  @Override
  public void sendMessageByQueue(String destinationName, String message) {
    logger.info(
        "KafkaClusterMessageSenderStrategy.sendMessage destinationName:{}, message:{}",
        destinationName,
        message);
    kafkaTemplate.send(destinationName, message);
  }

  @Override
  public void sendMessageByTopic(String topicName, String message) {}

  @Override
  public void sendMessageByTopic(String topicName, String message, String key) {

  }

  @Override
  public void sendMessageByQueue(
      String destinationName, MessageInteface<ActiveMqMessageBase> messageInteface) {
    // 消息业务唯一标识
    String bizId = messageInteface.getBizId();
    // 消息业务内容
    ActiveMqMessageBase bizBody = messageInteface.getBizBody();

    logger.info(
        "KafkaClusterMessageSenderStrategy.sendMessage bizId:{}, bizBody:{}", bizId, bizBody);

    kafkaTemplate.send(destinationName, messageInteface);
  }

  @Override
  public void sendMessageByTopic(
      String topicName, MessageInteface<ActiveMqMessageBase> messageInteface) {}
}
